// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: juntang

#include "coin2/exchange/apollox_futures/order_v1/parser.h"

#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/apollox_futures/order_v1/error_code.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::apollox_futures::order_v1 {

namespace {

using namespace coin2::exchange::apollox_futures::order_v1::error_code;

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::apollox::symbology::ApolloxCurrency;
using coin2::exchange::apollox_futures::symbology::ApolloxFuturesProduct;
using coin2::exchange::base::order::IOrderContext;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using TransferType = coin::proto::TransferRequestProto::TransferType;

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clientOrderId")) {
    std::string proc_str = data["clientOrderId"].GetString();
    try {
      proc_order_id = stoll(proc_str);
    } catch (std::exception& e) {
      LOG(ERROR) << e.what() << proc_str;
    }
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else if (type == "STOP_MARKET") {
    return coin::proto::OrderType::STOP_ORDER;
  } else if (type == "STOP") {
    return coin::proto::OrderType::STOP_LIMIT_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

void SetOrderStatus(const std::string& status, ProductOrderElement* order) {
  if (status == "REJECTED") {
    order->set_is_live(false);
  } else if (status == "NEW" || status == "CONFIRMED" || status == "PARTIALLY_FILLED") {
    order->set_is_live(true);
  } else if (status == "CANCELED" || status == "EXPIRED") {
    order->set_cancel_confirmed(true);
  } else if (status == "FILLED") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
}

bool IsApolloxFuturesErrMessage(const JsonValue& doc) {
  if (!doc.HasMember("code") || !doc.HasMember("msg")) {
    return false;
  }

  auto code = stoll(doc["code"].GetString());
  if (code != 200 || std::string_view(doc["msg"].GetString()) != "success") {
    return true;
  }

  return false;
}

bool IsAccountUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") && std::string_view(json["e"].GetString()) == "ACCOUNT_UPDATE";
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") && std::string_view(json["e"].GetString()) == "ORDER_TRADE_UPDATE";
}

}  // namespace

std::map<std::string, TransferType> ApolloxFuturesParser::transfer_type_map_{
    {"1", TransferRequestProto::WITHDRAW_SPOT},
    {"2", TransferRequestProto::DEPOSIT_SPOT},
    {"3", TransferRequestProto::WITHDRAW_SPOT},
    {"4", TransferRequestProto::DEPOSIT_SPOT}};

ApolloxFuturesParser::ApolloxFuturesParser() {}

void ApolloxFuturesParser::ParseRestAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

bool ApolloxFuturesParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  // {
  // 	"feeTier": 0,  		// account commisssion tier
  //  	"canTrade": true,  	// if can trade
  //  	"canDeposit": true,  	// if can transfer in asset
  //  	"canWithdraw": true, 	// if can transfer out asset
  //  	"updateTime": 0,
  //  	"totalInitialMargin": "0.00000000",    // total initial margin required with current mark
  //  price (useless with isolated positions), only for USDT asset 	"totalMaintMargin":
  //  "0.00000000",
  //  // total maintenance margin required, only for USDT asset 	"totalWalletBalance":
  //  "23.72469206",
  //  // total wallet balance, only for USDT asset 	"totalUnrealizedProfit": "0.00000000",   //
  //  total unrealized profit, only for USDT asset 	"totalMarginBalance": "23.72469206",     //
  //  total margin balance, only for USDT asset 	"totalPositionInitialMargin": "0.00000000",
  //  // initial margin required for positions with current mark price, only for USDT asset
  //  	"totalOpenOrderInitialMargin": "0.00000000",   // initial margin required for open orders
  //  with current mark price, only for USDT asset 	"totalCrossWalletBalance": "23.72469206", //
  //  crossed wallet balance, only for USDT asset 	"totalCrossUnPnl": "0.00000000",	  //
  //  unrealized profit of crossed positions, only for USDT asset 	"availableBalance":
  //  "23.72469206",
  //  // available balance, only for USDT asset 	"maxWithdrawAmount": "23.72469206"     //
  //  maximum amount for transfer out, only for USDT asset 	"assets": [
  //  		{
  //  			"asset": "USDT",			// asset name
  // 		   	"walletBalance": "23.72469206",      // wallet balance
  // 		   	"unrealizedProfit": "0.00000000",    // unrealized profit
  // 		   	"marginBalance": "23.72469206",      // margin balance
  // 		   	"maintMargin": "0.00000000",	    // maintenance margin required
  // 		   	"initialMargin": "0.00000000",    // total initial margin required with
  // current
  // mark price 		   	"positionInitialMargin": "0.00000000",    //initial margin
  // required for positions with current mark price 		   	"openOrderInitialMargin":
  // "0.00000000",   // initial margin required for open orders with current mark price
  // "crossWalletBalance": "23.72469206",
  // // crossed wallet balance 		   	"crossUnPnl": "0.00000000"       // unrealized
  // profit of crossed positions 		   	"availableBalance": "23.72469206",       //
  // available
  // balance 		   	"maxWithdrawAmount": "23.72469206",     // maximum amount for
  // transfer
  // out 		   	"marginAvailable": true,    // whether the asset can be used as
  // margin in Multi-Assets mode "updateTime": 1625474304765 // last update time
  // 		},
  // 		{
  //  			"asset": "BUSD",			// asset name
  // 		   	"walletBalance": "103.12345678",      // wallet balance
  // 		   	"unrealizedProfit": "0.00000000",    // unrealized profit
  // 		   	"marginBalance": "103.12345678",      // margin balance
  // 		   	"maintMargin": "0.00000000",	    // maintenance margin required
  // 		   	"initialMargin": "0.00000000",    // total initial margin required with
  // current
  // mark price 		   	"positionInitialMargin": "0.00000000",    //initial margin
  // required for positions with current mark price 		   	"openOrderInitialMargin":
  // "0.00000000",   // initial margin required for open orders with current mark price
  // "crossWalletBalance": "103.12345678",
  // // crossed wallet balance 		   	"crossUnPnl": "0.00000000"       // unrealized
  // profit of crossed positions 		   	"availableBalance": "103.12345678",       //
  // available
  // balance 		   	"maxWithdrawAmount": "103.12345678",     // maximum amount for
  // transfer
  // out 		   	"marginAvailable": true,    // whether the asset can be used as
  // margin in Multi-Assets mode "updateTime": 1625474304765 // last update time
  // 		}
  // 	],
  //  	"positions": [  // positions of all symbols in the market are returned
  //  		// only "BOTH" positions will be returned with One-way mode
  //  		// only "LONG" and "SHORT" positions will be returned with Hedge mode
  //  		{
  // 		 	"symbol": "BTCUSDT",  	// symbol name
  // 		   	"initialMargin": "0",	// initial margin required with current mark price
  // 		   	"maintMargin": "0",		// maintenance margin required
  // 		   	"unrealizedProfit": "0.00000000",  // unrealized profit
  // 		   	"positionInitialMargin": "0",      // initial margin required for positions
  // with current mark price 		   	"openOrderInitialMargin": "0",     // initial margin
  // required for open
  // orders with current mark price 		   	"leverage": "100",		// current
  // initial
  // leverage 		   	"isolated": true,  		// if the position is isolated
  // "entryPrice": "0.00000",  	// average entry price 		   	"maxNotional": "250000",
  // // maximum available notional with current leverage 		   	"positionSide":
  // "BOTH",  	// position side 		   	"positionAmt":
  // "0",			// position amount 		   	"updateTime": 0           //
  // last update time
  // 		}
  //   	]
  // }
  if (doc.IsObject() && IsApolloxFuturesErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("assets") || !doc["assets"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["assets"].GetArray()) {
    if (!data.HasMember("initialMargin") || !data.HasMember("marginBalance")) {
      continue;
    }
    double hold = stod(data["initialMargin"].GetString());
    double total = stod(data["marginBalance"].GetString());
    double avail = total - hold;
    CurrencyBalance balance;
    balance.set_currency_native(data["asset"].GetString());
    balance.set_available(avail);
    balance.set_hold(hold);
    balance.set_total(total);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }
  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance ApolloxFuturesParser::ParseBalance(const JsonValue& doc) {
  // {
  //   "asset": "USDT",
  //   "walletBalance": "0.41684018",
  //   "unrealizedProfit": "0.00754437",
  //   "marginBalance": "0.42438455",
  //   "maintMargin": "0.02959772",
  //   "initialMargin": "0.65108671",
  //   "positionInitialMargin": "0.29597721",
  //   "openOrderInitialMargin": "0.35510950",
  //   "maxWithdrawAmount": "0.41684018",
  //   "crossWalletBalance": "0.41684018",
  //   "crossUnPnl": "0.00754437",
  //   "availableBalance": "27.91019951",
  //   "marginAvailable": true,
  //   "updateTime": 1669363629254
  // },
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Apollox");
  acc_balance.set_market_type("Futures");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }

  return acc_balance;
}

void ApolloxFuturesParser::ParseRestAccountPosition() {
  JsonDocument doc;
  if (!PrepareJsonDocument(&doc)) {
    LOG(ERROR) << "[ParseRestAccountPosition] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_position = ParsePosition(doc);
  if (!account_position.has_exchange() || !account_position.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  result_.set_type(MsgType::ACCOUNT_POSITION);
  *result_.mutable_account_position() = std::move(account_position);
}

bool ApolloxFuturesParser::ParseRawPosition(
    const JsonValue& doc,
    RawPositionMap* raw_position_map) {
  if (!doc.IsArray()) {
    LOG(ERROR) << "[ParsePosition] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc.GetArray()) {
    if (!data.HasMember("symbol") || !data.HasMember("positionAmt") ||
        !data.HasMember("entryPrice")) {
      LOG(ERROR) << "Unexpected position data! " << JsonToString(data);
      continue;
    }
    auto native_symbol = data["symbol"].GetString();
    auto& position = (*raw_position_map)[native_symbol];
    double pos_amt = stod(data["positionAmt"].GetString());
    if (pos_amt >= 0.) {
      position.set_long_position(pos_amt);
    } else {
      position.set_short_position(-pos_amt);
    }
    position.set_net_position(pos_amt);
  }

  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountPosition ApolloxFuturesParser::ParsePosition(const JsonValue& doc) {
  RawPositionMap raw_position_map;
  auto success = ParseRawPosition(doc, &raw_position_map);
  if (!success) {
    return {};
  }

  AccountPosition acc_position;
  acc_position.set_exchange("Apollox");
  acc_position.set_market_type("Futures");
  for (const auto& native_symbol : symcache_->GetProductNative()) {
    if (raw_position_map.count(native_symbol) != 0) {
      continue;
    }
    auto* position = acc_position.add_each_position();
    position->set_long_position(0);
    position->set_short_position(0);
    position->set_net_position(0);
    position->set_available_long_position(0);
    position->set_available_short_position(0);
    position->set_native_symbol(native_symbol);

    DLOG(INFO) << "POS: " << native_symbol << " " << position->net_position();
  }

  for (auto& pair : raw_position_map) {
    auto* position = acc_position.add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position();
  }
  is_account_position_initialized_ = true;

  return acc_position;
}

bool ApolloxFuturesParser::ParseRawPositionFromWs(
    const JsonValue& doc,
    RawPositionMap* raw_position_map) {
  /*
  {
    "e": "ACCOUNT_UPDATE",				// Event Type
    "E": 1564745798939,            		// Event Time
    "T": 1564745798938 ,           		// Transaction
    "a":                          		// Update Data
      {
        "m":"ORDER",						// Event reason type
        "B":[                     		// Balances
          {
            "a":"USDT",           		// Asset
            "wb":"122624.12345678",    	// Wallet Balance
            "cw":"100.12345678",			// Cross Wallet Balance
            "bc":"50.12345678"			// Balance Change except PnL and Commission
          },
          {
            "a":"BUSD",
            "wb":"1.00000000",
            "cw":"0.00000000",
            "bc":"-49.12345678"
          }
        ],
        "P":[
          {
            "s":"BTCUSDT",          	// Symbol
            "pa":"0",               	// Position Amount
            "ep":"0.00000",            // Entry Price
            "cr":"200",             	// (Pre-fee) Accumulated Realized
            "up":"0",						// Unrealized PnL
            "mt":"isolated",				// Margin Type
            "iw":"0.00000000",			// Isolated Wallet (if isolated position)
            "ps":"BOTH"					// Position Side
          }，
          {
            "s":"BTCUSDT",
            "pa":"20",
            "ep":"6563.66500",
            "cr":"0",
            "up":"2850.21200",
            "mt":"isolated",
            "iw":"13200.70726908",
            "ps":"LONG"
          },
          {
            "s":"BTCUSDT",
            "pa":"-10",
            "ep":"6563.86000",
            "cr":"-45.04000000",
            "up":"-1423.15600",
            "mt":"isolated",
            "iw":"6570.42511771",
            "ps":"SHORT"
          }
        ]
      }
  }
  */
  if (!(doc.HasMember("E"))) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  if (!(doc.HasMember("a") && doc["a"].IsObject())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  if (!(doc["a"].HasMember("P") && doc["a"]["P"].IsArray())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }

  const auto& position_list = doc["a"]["P"];
  for (const auto& product_pos : position_list.GetArray()) {
    if (std::string_view(product_pos["ps"].GetString()) != "BOTH") {
      LOG(ERROR) << "Ws position ps not BOTH: " << JsonToString(doc);
      continue;
    }
    auto native_symbol = product_pos["s"].GetString();
    auto& position = (*raw_position_map)[native_symbol];
    auto net_pos = stod(product_pos["pa"].GetString());
    position.set_net_position(net_pos);
    if (net_pos >= 0) {
      position.set_long_position(net_pos);
    } else {
      position.set_short_position(-net_pos);
    }
  }

  return true;
}

bool ApolloxFuturesParser::ParseWsPosition(const JsonValue& doc, AccountPosition* acc_position) {
  acc_position->Clear();

  RawPositionMap raw_position_map;
  bool success = ParseRawPositionFromWs(doc, &raw_position_map);
  if (!success) {
    return false;
  }

  acc_position->set_exchange("Apollox");
  acc_position->set_market_type("Futures");
  for (auto& pair : raw_position_map) {
    auto* position = acc_position->add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position();
  }

  return true;
}

ProductOrderElement ApolloxFuturesParser::ParseRestOrderInfo(const JsonValue& data) {
  ProductOrderElement order;
  std::string native_symbol = data["symbol"].GetString();
  if (!symcache_->has_product_native(native_symbol)) {
    return order;
  }
  std::string exchange_id = fmt::format("{}{}", native_symbol, data["orderId"].GetString());
  int64_t order_timestamp_ft = msg_->timestamp;
  int64_t order_timestamp_et = stoll(data["time"].GetString());

  if (order_timestamp_ft / 1e9 - order_timestamp_et / 1e3 > 60 * 60) {
    auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
    if (oc != nullptr && !GetOcm()->IsInWorkingOrders(oc)) {
      // TODO(xguo): adjust query url not to list all previous orders. Otherwise too chatty.
      // LOG(INFO) << "Ignore apollox long time order: " << exchange_id;
      return order;
    }
  }
  order.set_market_type(MarketType::Futures);
  order.set_exchange_type(ExchangeType::Apollox);
  order.set_api_version("v1");
  order.set_exchange_order_id(exchange_id);
  order.set_symbol(native_symbol);
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["origQty"].GetString()));
  order.set_proc_order_id(GetProcOrderId(data));
  order.set_side(GetOrderSide(data["side"].GetString()));
  order.set_order_type(GetOrderType(data["type"].GetString()));
  order.set_timestamp(order_timestamp_ft);
  SetOrderStatus(data["status"].GetString(), &order);
  return order;
}

void ApolloxFuturesParser::ParseRestOpenOrders() {
  is_query_historical_orders_ = false;
  ParseRestOrderInfoList();
}

void ApolloxFuturesParser::ParseRestAllOrders() {
  is_query_historical_orders_ = true;
  ParseRestOrderInfoList();
}

void ApolloxFuturesParser::ParseRestOrderInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOpenOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsApolloxFuturesErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] " << JsonToString(data);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto new_order = ParseRestOrderInfo(info);
    // according to binance
    // Ignore open orders form QueryHistoricalOrder response
    // only trust open orders from QueryOpenOrders response.
    if (!(is_query_historical_orders_ && new_order.is_live()) && new_order.has_market_type()) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
      product_order_bundle_found = true;
    }
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle ApolloxFuturesParser::ParseWsOrderInfoList(const JsonValue& order_data) {
  // {
  //   "e":"ORDER_TRADE_UPDATE",		// Event Type
  //   "E":1568879465651,			// Event Time
  //   "T":1568879465650,			// Transaction Time
  //   "o":{
  //     "s":"BTCUSDT",				// Symbol
  //     "c":"TEST",					// Client Order Id
  //       // special client order id:
  //       // starts with "autoclose-": liquidation order
  //       // "adl_autoclose": ADL auto close order
  //     "S":"SELL",					// Side
  //     "o":"TRAILING_STOP_MARKET",	// Order Type
  //     "f":"GTC",					// Time in Force
  //     "q":"0.001",				// Original Quantity
  //     "p":"0",					// Original Price
  //     "ap":"0",					// Average Price
  //     "sp":"7103.04",				// Stop Price. Please ignore with
  //     TRAILING_STOP_MARKET order "x":"NEW",					// Execution Type
  //     "X":"NEW",					// Order Status
  //     "i":8886774,				// Order Id
  //     "l":"0",					// Order Last Filled Quantity
  //     "z":"0",					// Order Filled Accumulated Quantity
  //     "L":"0",					// Last Filled Price
  //     "N":"USDT",            	// Commission Asset, will not push if no commission
  //     "n":"0",               	// Commission, will not push if no commission
  //     "T":1568879465651,			// Order Trade Time
  //     "t":0,			        	// Trade Id
  //     "b":"0",			    	// Bids Notional
  //     "a":"9.91",					// Ask Notional
  //     "m":false,					// Is this trade the maker side?
  //     "R":false,					// Is this reduce only
  //     "wt":"CONTRACT_PRICE", 		// Stop Price Working Type
  //     "ot":"TRAILING_STOP_MARKET",	// Original Order Type
  //     "ps":"LONG",						// Position Side
  //     "cp":false,						// If Close-All, pushed with
  //     conditional order
  //     "AP":"7476.89",				// Activation Price, only puhed with
  //     TRAILING_STOP_MARKET order
  //     "cr":"5.0",					// Callback Rate, only puhed with
  //     TRAILING_STOP_MARKET order
  //     "rp":"0"							// Realized Profit of the
  //     trade
  // }
  ProductOrderBundle product_order_bundle;
  if (!order_data.IsObject()) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_order_bundle;
  }

  if (!order_data.HasMember("o")) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_order_bundle;
  }

  const auto& data = order_data["o"];
  auto* order = product_order_bundle.add_each_order();
  order->set_market_type(MarketType::Futures);
  order->set_exchange_type(ExchangeType::Apollox);
  order->set_api_version("v1");

  order->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
  order->set_symbol(data["s"].GetString());
  order->set_price(stod(data["p"].GetString()));
  order->set_qty(stod(data["q"].GetString()));
  order->set_proc_order_id(stoll(data["c"].GetString()));
  order->set_side(GetOrderSide(data["S"].GetString()));
  order->set_order_type(GetOrderType(data["o"].GetString()));
  SetOrderStatus(data["X"].GetString(), order);
  order->set_timestamp(msg_->timestamp);
  return product_order_bundle;
}

// {
// "e":"ORDER_TRADE_UPDATE",		// Event Type
// "E":1568879465651,			// Event Time
// "T":1568879465650,			// Transaction Time
// "o":{
//   "s":"BTCUSDT",				// Symbol
//   "c":"TEST",					// Client Order Id
//     // special client order id:
//     // starts with "autoclose-": liquidation order
//     // "adl_autoclose": ADL auto close order
//   "S":"SELL",					// Side
//   "o":"TRAILING_STOP_MARKET",	// Order Type
//   "f":"GTC",					// Time in Force
//   "q":"0.001",				// Original Quantity
//   "p":"0",					// Original Price
//   "ap":"0",					// Average Price
//   "sp":"7103.04",				// Stop Price. Please ignore with
//   TRAILING_STOP_MARKET order "x":"NEW",					// Execution Type
//   "X":"NEW",
//   // Order Status "i":8886774,				// Order Id "l":"0",
//   // Order Last Filled Quantity
//   "z":"0",					// Order Filled Accumulated Quantity
//   "L":"0",					// Last Filled Price
//   "N":"USDT",            	// Commission Asset, will not push if no commission
//   "n":"0",               	// Commission, will not push if no commission
//   "T":1568879465651,			// Order Trade Time
//   "t":0,			        	// Trade Id
//   "b":"0",			    	// Bids Notional
//   "a":"9.91",					// Ask Notional
//   "m":false,					// Is this trade the maker side?
//   "R":false,					// Is this reduce only
//   "wt":"CONTRACT_PRICE", 		// Stop Price Working Type
//   "ot":"TRAILING_STOP_MARKET",	// Original Order Type
//   "ps":"LONG",						// Position Side
//   "cp":false,						// If Close-All, pushed with
//   conditional order "AP":"7476.89",				// Activation Price, only puhed with
//   TRAILING_STOP_MARKET order "cr":"5.0",					// Callback Rate,
//   only puhed with TRAILING_STOP_MARKET order "rp":"0"
//   // Realized Profit of the trade
// }
ProductFillBundle ApolloxFuturesParser::ParseWsFillInfoList(const JsonValue& order_data) {
  ProductFillBundle product_fill_bundle;
  if (!order_data.HasMember("o")) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_fill_bundle;
  }
  const auto& data = order_data["o"];
  std::string_view status = data["X"].GetString();

  if (status == "FILLED" || status == "PARTIALLY_FILLED") {
    auto* fill = product_fill_bundle.add_each_fill();
    fill->set_symbol(data["s"].GetString());
    fill->set_fill_timestamp(msg_->timestamp);
    fill->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
    fill->set_proc_order_id(stoll(data["c"].GetString()));
    fill->set_price(stod(data["L"].GetString()));
    fill->set_qty(stod(data["l"].GetString()));
    fill->set_fill_id(data["t"].GetString());
    if (data["m"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (data.HasMember("N") && !data["N"].IsNull()) {
      const std::string fee_currency = data["N"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (data.HasMember("n") && !data["n"].IsNull()) {
          const std::string fee = data["n"].GetString();
          if (!fee.empty()) {
            fill->set_fee(-stod(fee));
          }
        }
      }
    }
  }
  return product_fill_bundle;
}

// [
//   {
//   	"buyer": false,
//   	"commission": "-0.07819010",
//   	"commissionAsset": "USDT",
//   	"id": 698759,
//   	"maker": false,
//   	"orderId": 25851813,
//   	"price": "7819.01",
//   	"qty": "0.002",
//   	"quoteQty": "15.63802",
//   	"realizedPnl": "-0.91539999",
//   	"side": "SELL",
//   	"positionSide": "SHORT",
//   	"symbol": "BTCUSDT",
//   	"time": 1569514978020
//   }
// ]
void ApolloxFuturesParser::ParseRestFillInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected rest fill msg!" << JsonToString(data);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(info["symbol"].GetString());
    fill->set_fill_timestamp(msg_->timestamp);
    fill->set_exchange_order_id(
        info["symbol"].GetString() + std::string(info["orderId"].GetString()));
    fill->set_price(stod(info["price"].GetString()));
    fill->set_qty(stod(info["qty"].GetString()));
    fill->set_fill_id(info["id"].GetString());
    if (info["maker"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (info.HasMember("commissionAsset") && !info["commissionAsset"].IsNull()) {
      const std::string fee_currency = info["commissionAsset"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (info.HasMember("commission") && !info["commission"].IsNull()) {
          const std::string fee = info["commission"].GetString();
          if (!fee.empty()) {
            fill->set_fee(-stod(fee));
          }
        }
      }
    }
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

std::string ApolloxFuturesParser::ParseHeader(
    const std::shared_ptr<AsioHttpContext>& http_context) {
  const auto& header = http_context->res.base();
  nlohmann::json json = {
      {"x-mbx-order-count-1d", -1},
      {"x-mbx-order-count-10s", -1},
      {"x-mbx-order-count-1m", -1},
      {"x-mbx-used-weight-1m", -1},
      {"retry-after", -1},
      {"bind_ip", http_context->bind_ip}};

  if (header.count("x-mbx-used-weight-1m") > 0) {
    json["x-mbx-used-weight-1m"] = std::stoi(header.at("x-mbx-used-weight-1m").data());
  }

  if (header.count("x-mbx-order-count-1m") > 0) {
    json["x-mbx-order-count-1m"] = std::stoi(header.at("x-mbx-order-count-1m").data());
  }

  if (header.count("x-mbx-order-count-10s") > 0) {
    json["x-mbx-order-count-10s"] = std::stoi(header.at("x-mbx-order-count-10s").data());
  }

  if (header.count("retry-after") > 0) {
    json["retry-after"] = std::stoi(header.at("retry-after").data());
  }
  return json.dump();
}

// {
//   "clientOrderId": "testOrder",
//   "cumQty": "0",
//   "cumQuote": "0",
//   "executedQty": "0",
//   "orderId": 22542179,
//   "avgPrice": "0.00000",
//   "origQty": "10",
//   "price": "0",
//     "reduceOnly": false,
//     "side": "BUY",
//     "positionSide": "SHORT",
//     "status": "NEW",
//     "stopPrice": "9300",		// please ignore when order type is TRAILING_STOP_MARKET
//     "closePosition": false,   // if Close-All
//     "symbol": "BTCUSDT",
//     "timeInForce": "GTC",
//     "type": "TRAILING_STOP_MARKET",
//     "origType": "TRAILING_STOP_MARKET",
//     "activatePrice": "9020",	// activation price, only return with TRAILING_STOP_MARKET order
//     "priceRate": "0.3",			// callback rate, only return with
//     TRAILING_STOP_MARKET order
//   "updateTime": 1566818724722,
//   "workingType": "CONTRACT_PRICE",
//   "priceProtect": false            // if conditional order trigger is protected
// }
// #error message
// {
// "code": -1121,
// "msg": "Invalid symbol."
// }
void ApolloxFuturesParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.base().result_int());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ApolloxFuturesSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[ApolloxFuturesSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ApolloxFuturesSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!json.IsObject() || IsApolloxFuturesErrMessage(json)) {
    LOG(ERROR) << "[ApolloxFutures Submit Order] Error Response" << http_context->res;
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    resp->set_success(false);
    return;
  }

  if (!json.HasMember("orderId")) {
    LOG(ERROR) << "[ApolloxFuturesSubmitOrder] exchange order id not found" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(
      json["symbol"].GetString() + std::string(json["orderId"].GetString()));
}

void ApolloxFuturesParser::ParseCancelResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.base().result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ApolloxFuturesCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[ApolloxFuturesCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ApolloxFuturesCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("symbol") && json.HasMember("orderId")) {
    resp->set_exchange_order_id(
        json["symbol"].GetString() + std::string(json["orderId"].GetString()));
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code =
        GetOrderErrorCode(json["code"], json["msg"], order_context->has_exchange_order_id());
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[ApolloxFuturesCancelOrder] Error Response" << http_context->res;
      resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
      resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
      resp->set_success(false);
      return;
    }
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);
}

void ApolloxFuturesParser::ParseTransfer() {
  /*
  {
    "rows":
    [
      {
        "asset": "USDT",
        "tranId": 100000001
        "amount": "40.84624400",
        "type": "1",
        "timestamp": 1555056425000,
        "status": "CONFIRMED"
      }
    ],
    "total": 1
  }
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    return;
  }

  if (data.IsObject() && IsApolloxFuturesErrMessage(data)) {
    LOG(ERROR) << "[ParseTransfer] " << JsonToString(data);
    return;
  }

  if (std::string_view(data["total"].GetString()) == "0") {
    LOG(INFO) << "[ParseTransfer] no transfer: " << JsonToString(data);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& row : data["rows"].GetArray()) {
    if (!row.HasMember("asset") || !row.HasMember("type") || !row.HasMember("amount")) {
      LOG(ERROR) << "[ParseTransfer] Invalid transfer format";
      continue;
    }
    if (!symcache_->has_currency_native(row["asset"].GetString())) {
      continue;
    }
    auto* currency = symcache_->GetCurrencyByNative(row["asset"].GetString());
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    std::string type = row["type"].GetString();
    if (transfer_type_map_.find(type) != transfer_type_map_.end()) {
      currency_transfer->set_type(transfer_type_map_[type]);
    }
    std::string status = row["status"].GetString();
    if (status == "CONFIRMED") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_transfer_id(row["tranId"].GetString());
    currency_transfer->set_qty(stod(row["amount"].GetString()));
    int64_t ts = std::stoll(row["timestamp"].GetString()) * 1e6;
    currency_transfer->set_create_timestamp(ts);
    if (ts > result_.last_transfer_timestamp_) {
      result_.last_transfer_timestamp_ = ts;
    }
    currency_transfer_bundle_found = true;
  }

  if (currency_transfer_bundle_found) {
    result_.set_type(BaseParser::MsgType::TRANSFER);
  }
}

void ApolloxFuturesParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (IsAccountUpdate(document)) {
    // make first rest position result before ws position result,
    // so SetAllZeroPosition won't take effect on partial symbol
    if (!is_account_position_initialized_) {
      result_.set_type(MsgType::IGNORABLE);
      return;
    }
    if (!ParseWsPosition(document, result_.mutable_account_position())) {
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      return;
    }
    // linchuan : ws balance lack of all symbol's unrealized profit, disabled.
    // result_.set_type(MsgType::ACCOUNT_BALANCE);
    result_.set_type(MsgType::ACCOUNT_POSITION);
  } else if (IsOrderUpdate(document)) {
    *result_.mutable_fill_list() = ParseWsFillInfoList(document);
    *result_.mutable_order_list() = ParseWsOrderInfoList(document);
    result_.set_type(MsgType::ORDER_STATUS);
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void ApolloxFuturesParser::ParseListenKey() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseListenKey] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  LOG(INFO) << "apollox/futures: " << JsonToString(document);
  if (document.HasMember("listenKey")) {
    std::string listen_key(document["listenKey"].GetString());
    mb_->PublishEvent("apollox/listen_key", &listen_key);
    last_key_query_ts_ = msg_->timestamp;
  } else {
    double delta_ts = (msg_->timestamp - last_key_query_ts_) / 1e9;
    LOG(ERROR) << "Failed to query listenKey" << JsonToString(document);
    LOG(INFO) << "key_age(seconds): " << delta_ts;
  }
}

auto ApolloxFuturesParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg_->type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  std::string s = ParseHeader(msg.http_context);
  mb_->PublishEvent("apollox/req_rate_limit", reinterpret_cast<void*>(&s));

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::LISTEN_KEY:
      ParseListenKey();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestAllOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_POSITION:
      ParseRestAccountPosition();
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }
  return result_;
}

}  // namespace coin2::exchange::apollox_futures::order_v1
